package com.hdaccp.ch04

import org.apache.spark.SparkConf
import org.apache.spark.streaming.flume.FlumeUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * spark streaming 入门
  *
  */
object Demo6 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
    val ssc = new StreamingContext(conf, Seconds(5))

    //val lines = ssc.socketTextStream("192.168.15.11", 1234)

    val flumeStream = FlumeUtils.createPollingStream(ssc, "192.168.15.11", 41413)

    flumeStream.map(x=> new String(x.event.getBody.array()).trim)
      .flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()

    ssc.start()             // Start the computation
    ssc.awaitTermination()


  }
}
